package com.we.risk.registerchanneldeprecated.browseinfo.deprecated;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

@Deprecated
public class AdmUserPhoneBrowseOriginDataDeprecated {
    public static final String INPUT_KEY_BY = "id_combine";
    public static final String RECV_TIME = "receive_time";
    public static final String UNIQUE_CHANNEL = "unique_channel";
    public static final String OUTPUT_KEY_BY = "uid";
    public static final String RELEASEPROP =
            "risk/registerchannel/kfk_user_phone_browse_origin_data_prod.properties";
    public static final String USERLABELPROP =
            "risk/registerchannel/adm_user_phone_browse_prod.properties";
    public static Logger LOG =
            LoggerFactory.getLogger(AdmUserPhoneBrowseOriginDataDeprecated.class);

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            String srckfkOffset = paramReader.getTmpKfkOffset();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty("auto.offset.reset", srckfkOffset);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            FlinkKafkaConsumer<String> kfkSource =
                    new FlinkKafkaConsumer<String>(
                            srcKafkaTopic, new SimpleStringSchema(), consumProp);
            DataStreamSource<String> input = env.addSource(kfkSource);

            SingleOutputStreamOperator<String> out =
                    input.uid("AdmBrowseOriginLabel-input")
                            .keyBy(
                                    new KeySelector<String, String>() {
                                        @Override
                                        public String getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getString(INPUT_KEY_BY);
                                        }
                                    })
                            .map(new UserlabelPartExpandRichMapFunc())
                            .filter(
                                    new FilterFunction<String>() {
                                        @Override
                                        public boolean filter(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            if (jsonObject.containsKey(OUTPUT_KEY_BY)) {
                                                return (jsonObject.get(OUTPUT_KEY_BY) != null);
                                            } else {
                                                return false;
                                            }
                                        }
                                    });
            //
            //            .keyBy(new KeySelector<String, String>() {
            //                @Override
            //                public String getKey(String value) throws Exception {
            //                    return JSONObject.parseObject(value).getString(OUTPUT_KEY_BY);
            //                }
            //            }).flatMap(new UserLabelRichFlatMapFunc());

            /** sink to Kafka */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> userLabelAllFlinkKafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            java.util.Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(userLabelAllFlinkKafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmUserPhoneBrowseOriginDataDeprecated.class.toString());
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class AdmBrowseOriginRichFlatMapFuncDeprecated
            extends RichFlatMapFunction<String, String> {
        MapState<Long, String> browseMState;
        ValueState<JSONObject> browseVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            StateTtlConfig ttlConfig =
                    StateTtlConfig.newBuilder(Time.hours(12))
                            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                            .cleanupInRocksdbCompactFilter(10000L)
                            .build();

            MapStateDescriptor<Long, String> browseMStateDesc =
                    new MapStateDescriptor<>("browseMState", Long.class, String.class);
            browseMStateDesc.enableTimeToLive(ttlConfig);
            browseMState = getRuntimeContext().getMapState(browseMStateDesc);
            // 保留最后一条点击数据
            browseVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "browseVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                // LOG.debug("input: " + value);
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    if (inputJson.containsKey(OUTPUT_KEY_BY)) {
                        // LOG.debug("input contains OUTPUT_KEY_BY ");
                        // user phone info 消息处理流程
                        long uid = inputJson.getLongValue(OUTPUT_KEY_BY);
                        // LOG.debug("uid: " + uid);
                        JSONObject outJson = new JSONObject();
                        outJson.put(OUTPUT_KEY_BY, uid);

                        if (!browseMState.isEmpty()) {
                            //
                            Iterator<Long> iterator = browseMState.keys().iterator();
                            while (iterator.hasNext()) {
                                Long recvTime = iterator.next();
                                String channel = browseMState.get(recvTime);
                                outJson.put(RECV_TIME, recvTime);
                                outJson.put(UNIQUE_CHANNEL, channel);

                                out.collect(outJson.toString());
                            }
                            // 清理已发送的记录
                            browseMState.clear();
                        } else {
                            // 更新State中的uid信息
                            // LOG.debug("browseMState is empty!");
                            JSONObject tmpState = browseVState.value();
                            JSONObject curState =
                                    (tmpState == null ? initJsonObjState(USERLABELPROP) : tmpState);
                            Long stateUid = curState.getLong(OUTPUT_KEY_BY);

                            // LOG.debug("stateUid: " + (stateUid == null ? "null" : stateUid));
                            if (stateUid == null || stateUid != uid) {
                                // 更新uid
                                curState.put(OUTPUT_KEY_BY, uid);
                                browseVState.update(curState);

                                // 发送最后一条点击信息
                                long last_recv_time = curState.getLongValue(RECV_TIME);
                                String last_channel = curState.getString(UNIQUE_CHANNEL);
                                // LOG.debug("last_recv_time: " + last_recv_time + " last_channel: "
                                // + last_channel);
                                if (last_channel != null && last_recv_time != 0) {
                                    outJson.put(RECV_TIME, last_recv_time);
                                    outJson.put(UNIQUE_CHANNEL, last_channel);
                                    out.collect(outJson.toString());
                                }
                            }
                        }
                    } else {
                        // browse广告点击流消息处理流程
                        JSONObject tmpState = browseVState.value();
                        JSONObject uidState =
                                (tmpState == null ? initJsonObjState(USERLABELPROP) : tmpState);
                        String uid = uidState.getString(OUTPUT_KEY_BY);
                        long state_recv_time = uidState.getLongValue(RECV_TIME);

                        long recvTime = inputJson.getLongValue(RECV_TIME);
                        String channel = inputJson.getString(UNIQUE_CHANNEL);

                        if (uid != null) {
                            // 关联到uid，直接发往下游
                            JSONObject outJson = new JSONObject();
                            outJson.put(RECV_TIME, recvTime);
                            outJson.put(UNIQUE_CHANNEL, channel);
                            outJson.put(OUTPUT_KEY_BY, uid);

                            out.collect(outJson.toString());
                        } else {
                            // 未关联到uid，保存MState
                            browseMState.put(recvTime, channel);

                            // 保存最后一条点击记录,以防ttl过期清理
                            if (recvTime > state_recv_time) {
                                uidState.put(RECV_TIME, recvTime);
                                uidState.put(UNIQUE_CHANNEL, channel);

                                browseVState.update(uidState);
                            }
                        }
                    }
                }
            } catch (Exception e) {
                LOG.error("Excep: " + e.toString());
                LOG.error("Excep input: " + value);
            }
        }
    }

    private static class UserlabelPartExpandRichMapFunc extends RichMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "userLabelVState", JSONObject.class));
        }

        @Override
        public String map(String value) throws Exception {

            JSONObject tmp = userLabelVState.value();
            JSONObject curJsonObj = (tmp == null ? initJsonObjState(USERLABELPROP) : tmp);
            JSONObject input = JSON.parseObject(value);
            for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                String key = inputEle.getKey();
                Object val = inputEle.getValue();
                if (curJsonObj.containsKey(key)) {
                    curJsonObj.put(key, val);
                }
                //                else {
                //                    LOG.error("Properties not contains key: " + key + " value: " +
                // val);
                //                }
            }
            userLabelVState.update(curJsonObj);
            return curJsonObj.toString();
            /** for Doris table */
            //            return curJsonObj.toString(SerializerFeature.WriteMapNullValue);
        }
    }

    private static class UserLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "userLabelVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            JSONObject tmp = userLabelVState.value();
            JSONObject curState = (tmp == null ? initJsonObjState(USERLABELPROP) : tmp);
            JSONObject input = JSON.parseObject(value);
            String userKeyVal = input.getString(OUTPUT_KEY_BY);
            for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                String key = inputEle.getKey();
                Object val = inputEle.getValue();

                if (!key.equals(OUTPUT_KEY_BY)) {
                    Object jsonVal = curState.get(key);
                    if (jsonVal != null && val != null) {
                        if (!jsonVal.equals(val)) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(key, val);
                            out.collect(jsonObject.toString());
                        }
                    } else {
                        if (jsonVal != val) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(key, val);
                            out.collect(jsonObject.toString());
                        }
                    }
                }
            }
            userLabelVState.update(curState);
        }
    }

    private static JSONObject initJsonObjState(String propFile) throws IOException {

        JSONObject res = new JSONObject();
        InputStream resourceAsStream =
                AdmUserPhoneBrowseOriginDataDeprecated.class
                        .getClassLoader()
                        .getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        Set<Object> propSets = reader.getProperties().keySet();
        for (Object key : propSets) {
            String typeValue = reader.get((String) key);
            String[] splits = typeValue.split(":");
            if (splits.length == 2) {
                String type = splits[0];
                String value = splits[1];
                jsonObjectSetValue(res, type, (String) key, value);
            } else {
                LOG.error("split Error: " + splits.length);
            }
        }
        return res;
    }

    private static void jsonObjectSetValue(
            JSONObject jsonObj, String type, String key, String value) {
        if (!value.equals("null")) {
            switch (type) {
                case "Long":
                    jsonObj.put(key, Long.parseLong(value));
                    break;
                case "String":
                    jsonObj.put(key, value);
                    break;
                case "Float":
                    jsonObj.put(key, Float.parseFloat(value));
                    break;
                case "Integer":
                    jsonObj.put(key, Integer.parseInt(value));
                    break;
                default:
                    LOG.error("Not support this type: " + type);
                    break;
            }
        } else {
            jsonObj.put(key, null);
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String key = jsonObject.getString(OUTPUT_KEY_BY);
            return key.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
